package com.atguigu.api

import java.util.Properties

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * @ClassName WindowTest
 * @Description
 * @Author Mr Yang
 * @Date 2020/9/2 22:57
 * @Version 1.0
 */
object WindowTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.getConfig.setAutoWatermarkInterval(100L)

    //使用kafka模拟流数据的输入
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

//    val stream = env.addSource( new FlinkKafkaConsumer011[String]("hellokafka", new SimpleStringSchema(), properties) )

    val stream = env.socketTextStream("localhost", 7777)
    //transform
    val dataStream = stream.map(
      data => {
        val dataArray = data.split(",")
        SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
      })
      //.assignAscendingTimestamps(_.timestamp*1000)
      //.assignTimestampsAndWatermarks( new MyAssigner )
      .assignTimestampsAndWatermarks( new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
      } )
    //统计10秒内的最小温度
    val minTempWindowStream = dataStream.map(
      data => (data.id, data.temperature) )
        .keyBy(_._1)
        .timeWindow( Time.seconds(10))  //开时间窗口
        .reduce((data1, data2) => (data1._1, data1._2.min(data2._2))) //用reduce做增量聚合

    minTempWindowStream.print("minTempWindowStream")
    dataStream.print("dataStream")
    env.execute("window test")
  }
}

//class MyAssigner() extends AssignerWithPeriodicWatermarks[SensorReading]{
//
//  val bound = 60*1000
//  var maxTs = Long.MinValue
//
//  override def getCurrentWatermark: Watermark = new Watermark(maxTs-bound)
//
//  override def extractTimestamp(t: SensorReading, l: Long): Long = {
//    maxTs = maxTs.max(t.timestamp * 1000)
//    t.timestamp * 1000
//  }
//}

class MyAssigner() extends AssignerWithPunctuatedWatermarks[SensorReading]{
  override def checkAndGetNextWatermark(t: SensorReading, l: Long): Watermark = new Watermark(l)

  override def extractTimestamp(t: SensorReading, l: Long): Long = t.timestamp * 1000
}